#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <sys/types.h>
#include "hdfs.h"

char *HOSTS[]={
	"cass",
	"compute-0-0.local",
	"compute-0-1.local",
	"compute-0-2.local",
	"compute-0-3.local",
	"compute-0-4.local",
	"compute-0-5.local",
	"compute-0-6.local",
	"compute-0-7.local",
	"compute-0-8.local",
	"compute-0-9.local",
	"compute-0-10.local",
	"compute-0-11.local",
	"compute-0-12.local",
	"compute-0-13.local",
	"compute-0-14.local",
	NULL
};

char *FILES[]={
	"/tmp/chombo3d/chombo3d_0.vti",
	"/tmp/chombo3d/chombo3d_1.vti",
	"/tmp/chombo3d/chombo3d_2.vti",
	"/tmp/chombo3d/chombo3d_3.vti",
	"/tmp/chombo3d/chombo3d_4.vti",
	"/tmp/chombo3d/chombo3d_5.vti",
	"/tmp/chombo3d/chombo3d_6.vti",
	"/tmp/chombo3d/chombo3d_7.vti",
	"/tmp/chombo3d/chombo3d_8.vti",
	"/tmp/chombo3d/chombo3d_9.vti",
	"/tmp/chombo3d/chombo3d_10.vti",
	"/tmp/chombo3d/chombo3d_11.vti",
	"/tmp/chombo3d/chombo3d_12.vti",
	"/tmp/chombo3d/chombo3d_13.vti",
	"/tmp/chombo3d/chombo3d_14.vti",
	"/tmp/chombo3d/chombo3d_15.vti",
	NULL
};

struct DCmonitor{
	int nr_hosts, nr_files, rep;
	char **host;
	char **file;
	std::vector< std::set< int > > fragments_node_contained;
	std::vector< std::set< int > > node_fragments_contained;
};

struct DCscheduler{
	int nr_results;
	istd::set< int > result;
	unsigned int current_hostid;
};

void DCscheduler_print(){
	printf("ok");
};


int DCscheduler_getHostID(struct DCmonitor *p, char *host)
{
	assert(p);
	assert(host);
	int i;
	for(i=0; i<p->nr_hosts; i++){
		if(!strcmp(p->host[i], host)){
			return i;
		}
	}
	assert(0);
	return -1;
}

struct DCmonitor *DCmonitor_create(char *host[], char *file[], int rep)
{
	assert(host);
	assert(file);

	struct DCmonitor *monitor=(struct DCmonitor *)malloc(sizeof(*monitor));
	assert(monitor);
	memset(monitor, 0, sizeof(*monitor));

	int i,len;
	/* the number of hosts */
	for(i=0;host[i];i++);
	monitor->nr_hosts=i;
	/* the number of files */
	for(i=0;file[i];i++);
	monitor->nr_files=i;
	monitor->rep=rep;

	/* alloc memory for host */
	monitor->host = (char **)malloc(sizeof(char *) * (monitor->nr_hosts+1));
	assert(monitor->host);
	memset(monitor->host, 0,sizeof(char *) * (monitor->nr_hosts+1));
	for(i=0; i<monitor->nr_hosts; i++){
		len = strlen(host[i]);
		monitor->host[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(monitor->host[i]);
		memset(monitor->host[i], 0, sizeof(char) * (len+1));
		strcpy(monitor->host[i], host[i]);
	}

	/* alloc memory for file */
	monitor->file = (char **)malloc(sizeof(char *) * (monitor->nr_files+1));
	assert(monitor->file);
	memset(monitor->file, 0, sizeof(char *) * (monitor->nr_files+1));
	for(i=0; i<monitor->nr_files; i++){
		len = strlen(file[i]);
		monitor->file[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(monitor->file[i]);
		memset(monitor->file[i], 0, sizeof(char) * (len+1));
		strcpy(monitor->file[i], file[i]);
	}

	/* //alloc memory for map (file --> hosts)
	monitor->map = (int **)malloc(sizeof(int *) * (monitor->nr_files+1));
	assert(monitor->map);
	memset(monitor->map, 0, sizeof(int *) * (monitor->nr_files+1));
	for(i=0; i<monitor->nr_files; i++){
		//len = monitor->nr_files;
		monitor->map[i] = (int *)malloc(sizeof(int) * (monitor->rep));
		assert(monitor->map[i]);
		memset(monitor->map[i], 0, sizeof(int) * (monitor->rep));
	}*/

	monitor->fragments_node_contained = vector<set<int>>(monitor->nr_files);
	assert(monitor->fragments_node_contained);
	
	monitor->node_fragments_contained = vector< set< int > >( monitor->nr_hosts);
	assert (monitor->node_fragments_contained);
		
	/* fill locations */
	hdfsFS fs = hdfsConnect("default", 0);
   	if(!fs) {
       	fprintf(stderr, "Oops! Failed to connect to hdfs!\n");
       	exit(-1);
   	} 

 	for(i=0; i<monitor->nr_files; i++){
		char ***location = hdfsGetHosts(fs, monitor->file[i], 0, 1);;
		int k;
		for(k=0;location[0][k];k++){
			int host_id = DCscheduler_getHostID(monitor, location[0][k]);
			int file_id = i;
			monitor->fragments_node_contained[file_id].insert( host_id);
			monitor->node_fragments_contained[host_id].insert(i);
		}
	}
	
	return monitor;
}

struct DCscheduler *DCscheduler_create(struct DCmonitor *monitor, unsigned int current_hostid)
{
	assert(monitor);
	
	struct DCscheduler *scheduler=(struct DCscheduler *)malloc(sizeof(*scheduler));
	assert(scheduler);
	memset(scheduler, 0, sizeof(*scheduler));


	/* alloc memory for result */
	int len = (monitor->nr_files)/(monitor->nr_hosts);
	scheduler->nr_results = len;
	if((monitor->nr_files)%(monitor->nr_hosts)!=0)
		scheduler->nr_results = len+1;

	scheduler->result = (int *)malloc(sizeof(int) * (len+1));
	assert(scheduler->result);
	memset(scheduler->result, 0, sizeof(int) * (len+1));

	scheduler->current_hostid = current_hostid;
	return scheduler;
}



void DCscheduler_Schedule(struct DCscheduler *scheduler, struct DCmonitor *monitor)
{

	assert(scheduler);
	assert(monitor);


	std::set <int> una_files;
	int f, nf, hostid;
	std::set< int >::const_iterator itern,iterf;

	for(f=0;f<monitor->nr_files;f++)
		una_files.insert(f);
	
	for (f=0; f<monitor->nr_files;f++)
		{
			hostid = f%monitor->nr_hosts;
			int most_balanced, fragmentnumber, maxblance;
			std::set< int >::const_iterator iter= monitor->node_fragments_contained[hostid].begin();

			if(iter != monitor->node_fragments_contained[hostid].end()) {
				most_balanced = -1; fragmentnumber = -1;
				for( ; iter != monitor->node_fragments_contained[hostid].end(); iter++ )
					{				
						maxblance = monitor->nr_files;
					
						itern = monitor->fragments_node_contained[ *iter].begin();

						for( ; itern != monitor->fragments_node_contained[ *iter].end(); itern++ ){
							if(hostid != *itern&& maxblance > monitor->node_fragments_contained[*itern].size())
							{
								maxblance= monitor->node_fragments_contained[*itern].size();
							}
						}
						if(maxblance > most_balanced)
						{
							most_balanced = maxblance;
							fragmentnumber = *iter;
						}				
					}
				}
			else
				{
					most_balanced = -1;
					  for (iterf = una_files.begin(); iterf!=una_files.end(); iterf++){
					  	
						maxblance = monitor->nr_files;
						itern = monitor->fragments_node_contained[ *iterf].begin();

						for( ; itern != monitor->fragments_node_contained[ *iterf].end(); itern++ ){
							if(hostid != *itern&& maxblance > monitor->node_fragments_contained[*itern].size())
							{
								maxblance= monitor->node_fragments_contained[*itern].size();
							}
						}
						if(maxblance > most_balanced)
						{
							most_balanced = maxblance;
							fragmentnumber = *iterf;
						}
					  }
				}
			
			if(scheduler->current_hostid==hostid){
				scheduler->result.insert( fragmentnumber);
				}

			std::set< int >::const_iterator iterd = monitor->fragments_node_contained[fragmentnumber].begin();
					
			for( ; iterd != monitor->fragments_node_contained[fragmentnumber].end(); iterd++ )
				monitor->node_fragments_contained[*iterd].erase(fragmentnumber);
			una_files.erase(fragmentnumber);
	}
}


void DCscheduler_SimpleSchedule(struct DCscheduler *obj, struct DCmonitor *monitor)
{
	assert(obj);
	assert(monitor);
	
	int len = monitor->nr_hosts;
	int* hostscore = (int *) malloc (sizeof(int) * len);
	assert(hostscore);
	memset(hostscore, 0, sizeof(int) * len);

	/* algorithm */
	int fileid, hostid, i,j, minscore;
	int k =0;
	for(i=0; i<monitor->nr_files; i++){
		minscore = monitor->nr_files;
		for(j=0;j<monitor->rep;j++)
			if(hostscore[monitor->map[i][j]] < minscore){
				minscore = hostscore[monitor->map[i][j]];
				hostid = monitor->map[i][j];
			}
		hostscore[hostid]++;
		if(obj->current_hostid==hostid)
			obj->result[k++]=i;
	}
}

/*
int DCscheduler_check(struct DCscheduler *obj, int fileid)
{
	assert(obj);
	//fix later
	//assert((0<=fileid) && (fileid<obj->nr_files));
	int hostid = obj->current_hostid;
	printf("CHECK[%d][%d]=%d\n", obj->current_hostid, fileid, obj->result[hostid][fileid]);
	return obj->result[hostid][fileid];
}
*/
void DCscheduler_showmap(struct DCmonitor *obj, char *files[])
{
	assert(obj);
	int h,f;
	printf("********SHOW MAP ***********\n");
	for(f=0; f<obj->nr_files; f++){
		printf("file %s:\t", files[f]);
		for(h=0; h<obj->rep; h++){
			printf("\t%d", obj->map[f][h]);
		}
		printf("\n");
	}
}

void DCscheduler_showresult(struct DCscheduler *scheduler)
{
	assert(scheduler);
	int h,f;
	printf("******** %d SHOW RESULT of %d********\n", scheduler->nr_results, scheduler->current_hostid);
	for(f=0; f<scheduler->nr_results; f++){
		printf("\t%d", scheduler->result[f]);	
	}
	printf("\n");
}


void DCscheduler_destroy(struct DCscheduler *obj)
{
	assert(obj);
	int i;
	/* free result */
	free(obj->result);
	free(obj);
}

void DCmonitor_destroy(struct DCmonitor *obj)
{
	/* free map */
	int i;
	for(i=0; i<obj->nr_files; i++){
		free(obj->map[i]);
	}
	free(obj->map);

	/* free file */
	for(i=0; i<obj->nr_files; i++){
		free(obj->file[i]);
	}
	free(obj->file);

	/* free host */
	for(i=0; i<obj->nr_hosts; i++){
		free(obj->host[i]);
	}
	free(obj->host);

	/* free obj */
	free(obj);
}


#ifdef SCHEDULER1
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>
#include <sys/types.h>
#include "hdfs.h"
#include "DCScheduler.h"

char *HOSTS[]={
	"cass",
	"compute-0-0.local",
	"compute-0-1.local",
	"compute-0-2.local",
	"compute-0-3.local",
	"compute-0-4.local",
	"compute-0-5.local",
	"compute-0-6.local",
	"compute-0-7.local",
	"compute-0-8.local",
	"compute-0-9.local",
	"compute-0-10.local",
	"compute-0-11.local",
	"compute-0-12.local",
	"compute-0-13.local",
	"compute-0-14.local",
	NULL
};

char *FILES[]={
	"/tmp/chombo3d/chombo3d_0.vti",
	"/tmp/chombo3d/chombo3d_1.vti",
	"/tmp/chombo3d/chombo3d_2.vti",
	"/tmp/chombo3d/chombo3d_3.vti",
	"/tmp/chombo3d/chombo3d_4.vti",
	"/tmp/chombo3d/chombo3d_5.vti",
	"/tmp/chombo3d/chombo3d_6.vti",
	"/tmp/chombo3d/chombo3d_7.vti",
	"/tmp/chombo3d/chombo3d_8.vti",
	"/tmp/chombo3d/chombo3d_9.vti",
	"/tmp/chombo3d/chombo3d_10.vti",
	"/tmp/chombo3d/chombo3d_11.vti",
	"/tmp/chombo3d/chombo3d_12.vti",
	"/tmp/chombo3d/chombo3d_13.vti",
	"/tmp/chombo3d/chombo3d_14.vti",
	"/tmp/chombo3d/chombo3d_15.vti",
	NULL
};

void DCscheduler_print(){
	printf("ok");
};

int DCscheduler_getHostID(struct DCscheduler *p, char *host)
{
	assert(p);
	assert(host);
	int i;
	for(i=0; i<p->nr_hosts; i++){
		if(!strcmp(p->host[i], host)){
			return i;
		}
	}
	assert(0);
	return -1;
}



struct DCscheduler *DCscheduler_create(char *host[], char *file[], int current_hostid)
{
	assert(host);
	assert(file);

	printf("hostid=%d, %s\n", current_hostid, host[current_hostid]);

	struct DCscheduler *obj=(struct DCscheduler *)malloc(sizeof(*obj));
	assert(obj);
	memset(obj, 0, sizeof(*obj));

	int i,len;
	/* the number of hosts */
	for(i=0;host[i];i++);
	obj->nr_hosts=i;
	/* the number of files */
	for(i=0;file[i];i++);
	obj->nr_files=i;

	/* alloc memory for host */
	obj->host = (char **)malloc(sizeof(char *) * (obj->nr_hosts+1));
	assert(obj->host);
	memset(obj->host, 0,sizeof(char *) * (obj->nr_hosts+1));
	for(i=0; i<obj->nr_hosts; i++){
		len = strlen(host[i]);
		obj->host[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(obj->host[i]);
		memset(obj->host[i], 0, sizeof(char) * (len+1));
		strcpy(obj->host[i], host[i]);
	}

	/* alloc memory for file */
	obj->file = (char **)malloc(sizeof(char *) * (obj->nr_files+1));
	assert(obj->file);
	memset(obj->file, 0, sizeof(char *) * (obj->nr_files+1));
	for(i=0; i<obj->nr_files; i++){
		len = strlen(file[i]);
		obj->file[i] = (char *)malloc(sizeof(char) * (len+1));
		assert(obj->file[i]);
		memset(obj->file[i], 0, sizeof(char) * (len+1));
		strcpy(obj->file[i], file[i]);
	}

	/* alloc memory for map */
	obj->map = (unsigned char **)malloc(sizeof(unsigned char *) * (obj->nr_hosts+1));
	assert(obj->map);
	memset(obj->map, 0, sizeof(unsigned char *) * (obj->nr_hosts+1));
	for(i=0; i<obj->nr_hosts; i++){
		len = obj->nr_files;
		obj->map[i] = (unsigned char *)malloc(sizeof(unsigned char) * (len+1));
		assert(obj->map[i]);
		memset(obj->map[i], 0, sizeof(unsigned char) * (len+1));
	}


	/* fill locations */
	hdfsFS fs = hdfsConnect("default", 0);
   	if(!fs) {
       	fprintf(stderr, "Oops! Failed to connect to hdfs!\n");
       	exit(-1);
   	} 

 	for(i=0; i<obj->nr_files; i++){
		char ***location = hdfsGetHosts(fs, obj->file[i], 0, 1);;
		int k;
		for(k=0;location[0][k];k++){
			int host_id = DCscheduler_getHostID(obj, location[0][k]);
			int file_id = i;
			obj->map[host_id][file_id] = 1;
		}
	}
	

	/* alloc memory for result */
	obj->result = (unsigned char **)malloc(sizeof(unsigned char *) * (obj->nr_hosts+1));
	assert(obj->result);
	memset(obj->result, 0, sizeof(unsigned char *) * (obj->nr_hosts+1));
	for(i=0; i<obj->nr_hosts; i++){
		len = obj->nr_files;
		obj->result[i] = (unsigned char *)malloc(sizeof(unsigned char) * (len+1));
		assert(obj->result[i]);
		memset(obj->result[i], 0, sizeof(unsigned char) * (len+1));
	}
	obj->current_hostid = current_hostid;
	return obj;
}


static int DCscheduler_computeFileScore(struct DCscheduler *obj, int fileid)
{
	assert(obj);
	assert((0<=fileid) && (fileid<obj->nr_files));

	int score=0;
	int i;
	for(i=0; i<obj->nr_hosts; i++){
		if(obj->map[i][fileid]){
			score++;
		}
	}
	return score;
}
static int DCscheduler_pickfile(struct DCscheduler *obj)
{
	assert(obj);
	int min_score=obj->nr_hosts+1;
	int min_fileid=-1;
	int i;
	for(i=0; i<obj->nr_files; i++){
		int score = DCscheduler_computeFileScore(obj, i);
		if(score==0) continue;
		if(score < min_score){
			min_score = score;
			min_fileid = i;
		}
	}
	return min_fileid;
}
void DCscheduler_visIOSchedule(struct DCscheduler *obj)
{
	assert(obj);
	/* alloc memory for fileAssigned */
	unsigned char *fileAssigned=(unsigned char *)malloc(obj->nr_files);
	assert(fileAssigned);
	memset(fileAssigned, 0, obj->nr_files);
	/* alloc memory for hostAssigned */
	unsigned char *hostAssigned=(unsigned char *)malloc(obj->nr_hosts);
	assert(hostAssigned);
	memset(hostAssigned, 0, obj->nr_hosts);

	/* algorithm */
	int fileid, hostid, i,j;
	while((fileid=DCscheduler_pickfile(obj))!=-1){
		/* save the result */
		for(i=0; i<obj->nr_hosts; i++){
			if(obj->map[i][fileid]){
				hostid = i;
				obj->result[hostid][fileid] = 1;
				fileAssigned[fileid]=1;
				hostAssigned[hostid]=1;
				break;
			}
		}
		assert(i<obj->nr_hosts);
		/* eliminate the fileid in map */
		for(i=0; i<obj->nr_hosts; i++){
			obj->map[i][fileid]=0;
		}
		/* eliminate the hostid in map */
		for(i=0; i<obj->nr_files; i++){
			obj->map[hostid][i]=0;
		}
	}
	/* find files that have not been assigned */
	for(i=0; i<obj->nr_files; i++){
		if(fileAssigned[i]) continue;
		fileid = i;
		/* file i is not assigned */
		for(j=0; j<obj->nr_hosts; j++){
			if(hostAssigned[j]) continue;
			hostid = j;
			break;
		}
		if(j==obj->nr_hosts) break;
		obj->result[hostid][fileid]=1;
		fileAssigned[fileid]=1;
		hostAssigned[hostid]=1;
	}
	free(hostAssigned);
	free(fileAssigned);
}

static int DCscheduler_computeHostScore(struct DCscheduler *obj, int hostid)
{
	assert(obj);
	assert((0<=hostid) && (hostid<obj->nr_hosts));
	int score=0;
	int i;
	for(i=0; i<obj->nr_files; i++){
		if(obj->result[hostid][i]){
			score++;
		}
	}
	return score;
}
static int DCscheduler_pickhost(struct DCscheduler *obj, int fileid)
{
	assert(obj);
	assert((0<=fileid) && (fileid<obj->nr_files));
	int hostid;
	int min_score=obj->nr_files+1;
	int min_hostid=-1;
	int i;
	for(i=0; i<obj->nr_hosts; i++){
		if(obj->map[i][fileid]){
			hostid = i;
			/* hostid contains one replica */
			int score=DCscheduler_computeHostScore(obj,hostid);
			if(score < min_score){
				min_score = score;
				min_hostid = hostid;
			}
		}
	}
	return min_hostid;
}



void DCscheduler_Schedule(struct DCscheduler *obj)
{
	assert(obj);

	/* algorithm */
	int fileid, hostid, i,j;
	for(i=0; i<obj->nr_files; i++){
		fileid = i;
		/* search the fileid's replica */
		hostid = DCscheduler_pickhost(obj,fileid);
		assert((0<=hostid) && (hostid<obj->nr_hosts));
		/* assign the fileid to the hostid */
		obj->result[hostid][fileid] = 1;
	}
}

void DCscheduler_SimpleSchedule(struct DCscheduler *obj)
{
	assert(obj);

	/* algorithm */
	int fileid, hostid, i,j;
	for(i=0; i<obj->nr_files; i++){
		fileid = i;
		/* search the fileid's replica */
		hostid = DCscheduler_pickhost(obj,fileid);
		assert((0<=hostid) && (hostid<obj->nr_hosts));
		/* assign the fileid to the hostid */
		obj->result[hostid][fileid] = 1;
	}
}

int DCscheduler_check(struct DCscheduler *obj, int fileid)
{
	assert(obj);
	assert((0<=fileid) && (fileid<obj->nr_files));
	int hostid = obj->current_hostid;
	printf("CHECK[%d][%d]=%d\n", obj->current_hostid, fileid, obj->result[hostid][fileid]);
	return obj->result[hostid][fileid];
}
void DCscheduler_showmap(struct DCscheduler *obj)
{
	assert(obj);
	int h,f;
	printf("******** SHOW MAP ***********\n");
	for(f=0; f<obj->nr_files; f++){
		printf("file %4d:\t", f);
		for(h=0; h<obj->nr_hosts; h++){
			printf("\t%d", obj->map[h][f]);
		}
		printf("\n");
	}
}

void DCscheduler_showresult(struct DCscheduler *obj)
{
	assert(obj);
	int h,f;
	printf("******** SHOW RESULT ********\n");
	for(f=0; f<obj->nr_files; f++){
		printf("file %4d:\t", f);
		for(h=0; h<obj->nr_hosts; h++){
			printf("\t%d", obj->result[h][f]);
		}
		printf("\n");
	}
}
void DCscheduler_destroy(struct DCscheduler *obj)
{
	assert(obj);
	int i;
	/* free result */
	for(i=0; i<obj->nr_hosts; i++){
		free(obj->result[i]);
	}
	free(obj->result);

	/* free map */
	for(i=0; i<obj->nr_hosts; i++){
		free(obj->map[i]);
	}
	free(obj->map);

	/* free file */
	for(i=0; i<obj->nr_files; i++){
		free(obj->file[i]);
	}
	free(obj->file);

	/* free host */
	for(i=0; i<obj->nr_hosts; i++){
		free(obj->host[i]);
	}
	free(obj->host);

	/* free obj */
	free(obj);
}

#endif

/*
int main(int argc, char *argv[])
{
	printf("Hello World!\n");
	struct DCscheduler *obj = DCscheduler_create(HOSTS, FILES,0);
	DCscheduler_showmap(obj);
	DCscheduler_SimpleSchedule(obj);
	DCscheduler_showresult(obj);
	DCscheduler_destroy(obj);
	return 0;
}
*/
